package com.lagou.gkk;

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

public class Demo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);
        DataStreamSource<Qiye> data = env.fromElements(
                new Qiye(1l, "create", 1597905234000L),
                new Qiye(1l, "create", 1597905235000L),
                new Qiye(1l, "create", 1597905236000L),
                new Qiye(1l, "create", 1597905237000L),
                new Qiye(1l, "create", 1597905238000L),
                new Qiye(1l, "create", 1597905239000L),
                new Qiye(2l, "create", 1597905240000L),
                new Qiye(2l, "ok", 1597905237000L),
                new Qiye(3l, "create", 1597905239000L),
                new Qiye(4l, "create", 1597905238000L)
        );
        SingleOutputStreamOperator<Qiye> watermarks = data.assignTimestampsAndWatermarks(new WatermarkStrategy<Qiye>() {
            @Override
            public WatermarkGenerator<Qiye> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                return new WatermarkGenerator<Qiye>() {
                    long maxTimeStamp = Long.MIN_VALUE;
                    long maxOutOfOrderness = 500l;

                    @Override
                    public void onEvent(Qiye event, long eventTimestamp, WatermarkOutput output) {
                        maxTimeStamp = Math.max(maxTimeStamp, event.getTs());
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {
                        output.emitWatermark(new Watermark(maxTimeStamp - maxOutOfOrderness));
                    }
                };
            }
        }.withTimestampAssigner((element, recordTimestamp) -> element.getTs()));
        KeyedStream<Qiye, Long> keyed = watermarks.keyBy(value -> value.getId());

        Pattern<Qiye, Qiye> pattern = Pattern.<Qiye>begin("start").where(new IterativeCondition<Qiye>() {
            @Override
            public boolean filter(Qiye value, Context<Qiye> ctx) throws Exception {
                return value.getState().equals("create");
            }
        })
                .timesOrMore(5)
                .within(Time.seconds(60 * 2));

        PatternStream<Qiye> patternStream = CEP.pattern(keyed, pattern);
        OutputTag<Qiye> outputTag = new OutputTag<Qiye>("outoftime") {
        };
       /* SingleOutputStreamOperator<Qiye> result = patternStream.select(outputTag, new PatternTimeoutFunction<Qiye, Qiye>() {
            @Override
            public Qiye timeout(Map<String, List<Qiye>> pattern, long timeoutTimestamp) throws Exception {
                return pattern.get("start").get(0);
            }
        }, new PatternSelectFunction<Qiye, Qiye>() {
            @Override
            public Qiye select(Map<String, List<Qiye>> pattern) throws Exception {
                return pattern.get("start").get(0);
            }
        });
        DataStream<Qiye> sideOutput = result.getSideOutput(outputTag);
        sideOutput.print();
        */

        SingleOutputStreamOperator<Qiye> result = patternStream.process(new PatternProcessFunction<Qiye, Qiye>() {
            @Override
            public void processMatch(Map<String, List<Qiye>> match, Context ctx, Collector<Qiye> out) throws Exception {
                System.out.println(match);
//                out.collect(match.get("start").get(0));
            }
        });
        result.print();
        env.execute();
    }
}
